package com.rocketmq.consumer.handle.impl;

import com.rocketmq.consumer.handle.AbstractListenerHandler;
import com.rocketmq.consumer.service.ConsumerTest;
import com.rocketmq.consumer.utils.StatisticCount;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

/**
 * @Description:
 * @author: zhengfei
 * @param:
 * @date: 2018-06-22 下午6:26
 */

@Component("listenerHandlerOne")
public class ListenerHandlerOne extends AbstractListenerHandler {
    @Autowired
    private ConsumerTest consumerTest;
    private StatisticCount statisticCounter;

    @Override
    public void handle(String body) throws Exception {
        if (StringUtils.isBlank(body)) {
            return;
        }
        Thread.sleep(5000);
        consumerTest.process(body, "one");
        addStatisticCount(1,"one");
    }

    @Value("${testMqTopic}")
    @Override
    public void setTopicName(String topicName) {
        this.topicName = topicName;

    }

    @Value("${testMqTopicConsumerOne}")
    @Override
    public void setConsumerName(String consumerName) {
        this.consumerName = consumerName;
    }

    @Override
    public void setTagsName(String tagsName) {

    }

    /** 统计kafka发送记录日志*/
    private void addStatisticCount(int num , String type) {
        if (statisticCounter == null) {
            statisticCounter = StatisticCount.getCounter(type);
        }
        statisticCounter.addCount(num);
    }
}
